package com.inspurs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class DataClean {

    public static class Mapper_DataClean extends Mapper<LongWritable, Text, NullWritable, Text> {

        Map<String,String> map=new HashMap<String,String>();
        List<String> westList = new ArrayList<String>();
        List<String> eastList = new ArrayList<String>();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            super.setup(context);

            eastList.add("纽约尼克斯队");
            eastList.add("布鲁克林篮网队");
            eastList.add("波士顿凯尔特人队");
            eastList.add("多伦多猛龙队");
            eastList.add("费城76人队");
            eastList.add("底特律活塞队");
            eastList.add("印第安纳步行者队");
            eastList.add("克里夫兰骑士队");
            eastList.add("密尔沃基雄鹿队");
            eastList.add("芝加哥公牛队");
            eastList.add("迈阿密热火队");
            eastList.add("奥兰多魔术队");
            eastList.add("华盛顿奇才队");
            eastList.add("亚特兰大老鹰队");
            eastList.add("夏洛特黄蜂队");

            westList.add("明尼苏达森林狼队");
            westList.add("丹佛掘金队");
            westList.add("犹他爵士队");
            westList.add("波特兰开拓者队");
            westList.add("俄克拉荷马雷霆队");
            westList.add("萨克拉门托国王队");
            westList.add("洛杉矶湖人队");
            westList.add("菲尼克斯太阳队");
            westList.add("金州勇士队");
            westList.add("洛杉矶快船队");
            westList.add("圣安东尼奥马刺队");
            westList.add("达拉斯独行侠队");
            westList.add("休斯敦火箭队");
            westList.add("孟菲斯灰熊队");
            westList.add("新奥尔良鹈鹕队");

            map.put("费城勇士队","金州勇士队");
            map.put("旧金山勇士队","金州勇士队");
            map.put("明尼阿波利斯湖人队","洛杉矶湖人队");
            map.put("塞拉库斯民族队","费城76人队");
            map.put("罗切斯特皇家队","萨克拉门托国王队");
            map.put("圣路易斯老鹰队","亚特兰大老鹰队");
            map.put("华盛顿子弹队","华盛顿奇才队");
            map.put("巴尔的摩子弹队","华盛顿奇才队");
            map.put("西雅图超音速队","俄克拉荷马城雷霆队");
            map.put("福特韦恩活塞队","底特律活塞队");
            map.put("新泽西网队","布鲁克林篮网队");
            map.put("达拉斯小牛队","达拉斯独行侠队");

        }

        public String strArrToString(String[] str) {
            StringBuilder sBuilder = new StringBuilder("");
            for(int i = 0; i < str.length; i++) {
                if(i == 0)
                    sBuilder.append(str[i]);
                else
                    sBuilder.append("," + str[i]);
            }
            return sBuilder.toString();
        }

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] split = value.toString().split(",");

            String champion = split[2];
            String second = split[4];

            if(map.containsKey(champion))
                champion = map.get(champion);

            if(map.containsKey(second))
                second = map.get(second);

            split[2] = champion;
            split[4] = second;

            int year = Integer.parseInt(split[0]);
            Text te = new Text();
            if(year >= 1970) {
                String flag = "";
                if (eastList.contains(champion))
                    flag = "E";
                if (westList.contains(champion))
                    flag = "W";
                te.set(strArrToString(split) + "," + flag);
            } else {
                te.set(strArrToString(split));
            }
            context.write(NullWritable.get(), te);

        }

    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();

        //get job

        Job job = Job.getInstance(conf,"data clean");

        //Set up the run main class for the job
        job.setJarByClass(DataClean.class);

        //设置map阶段
        job.setMapperClass(DataClean.Mapper_DataClean.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop01:8020/0622/data/nba"));

        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop01:8020/0622/mapreduce_nba/clear_data"));

        int isok = job.waitForCompletion(true) ? 0 : 1;

        //退出job
        System.exit(isok);
    }

}
